from pyspark import *

if __name__ == '__main__':
    conf = SparkConf().setAppName("1_mapValues").setMaster("local[*]")
    sc = SparkContext(conf=conf)
    rdd1 = sc.parallelize([("hadoop", 1), ("java", 2), ("python", 3), ("python", 10)])
    rdd2 = sc.parallelize(["hadoop", "java", "python", "python"])
    rdd3 = sc.parallelize([("hadoop", 1)])

    print(rdd1.reduceByKey(lambda a,b:a+b).collect())
    # 察看每个分区的数据
    print(rdd1.glom().collect())
    # 只针对二元元组
    print(rdd1.mapValues(lambda v: v * 10).collect())
    # 不同类型的也可以合并
    print(rdd1.union(rdd2).collect())
    print(rdd1.groupBy(lambda t: t[0]).map(lambda t: (t[0], list(t[1]))).collect())
    print(rdd1.groupByKey().collect())
    # True 保留 False丢弃
    print(rdd1.filter(lambda t: t[1] % 2 == 1).collect())
    print(rdd1.mapValues(lambda x:x*10).collect())
    print(rdd1.intersection(rdd3).collect())
    print(rdd1.join(rdd3).collect())
    print(rdd1.leftOuterJoin(rdd3).collect())
    sc.stop()
